package com.boot2.core.mqtt;

import com.alibaba.fastjson.JSONObject;
import com.boot2.core.cache.redis.RedisDistributedLock;
import com.boot2.core.utils.RandomUtils;
import com.boot2.core.utils.StringUtils;
import lombok.extern.apachecommons.CommonsLog;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * @author zhangweilin
 * @description: mqtt消息分布式锁
 * @date 2019/2/28
 */
@Component
//@ConditionalOnBean({RedisConnectionFactory.class})
@ConditionalOnProperty("spring.redis.host")
@CommonsLog
public class MqttMessageLocker {

    @Autowired(required = false)
    RedisDistributedLock redisDistributedLock;

    @Autowired
    MqttProperties mqttProperties;

    Class<AbstractMqttMessage> mqttCLass = null;

    @PostConstruct
    public void init() {
        String mqttMessageClassName = mqttProperties.getMqttMessageClassName();
        if (StringUtils.isEmptyString(mqttMessageClassName)) {
            return;
        }
//        Assert.isTrue(StringUtils.isNotEmpty(mqttMessageClassName), "mqtt没有配置mqtt.mqttMessageClassName, 直接返回false");
        try {
            mqttCLass = (Class<AbstractMqttMessage>) Class.forName(mqttMessageClassName);
        } catch (ClassNotFoundException e) {
            e.printStackTrace();
            log.error("找不到类 " + mqttMessageClassName + " , 直接返回false");
        }
    }

    /**
     * 为 接收到的 mqtt消息获取分布式锁
     *
     * @param prefix
     * @param message
     * @return
     */
    public boolean getDistLock(String prefix, Message<?> message) {
        if (mqttCLass == null) {
            return false;
        }
        Object payload = message.getPayload();
        AbstractMqttMessage abstractMqttMessage = null;
        if (payload instanceof String) {
            String json = (String) payload;
            try {
                abstractMqttMessage = JSONObject.parseObject(json, mqttCLass);
            } catch (Exception e) {
                e.printStackTrace();
                log.error("mqtt业务消息, 转换失败,json: " + json, e);
                return false;
            }
        }
        String key = abstractMqttMessage.getMsgId();
        String deviceSn = abstractMqttMessage.getDeviceSn();
        if (StringUtils.isEmpty(key)) {
            key = payload.toString();
        }
        key = prefix + deviceSn + "_" + key;
        log.info("mqtt message key:" + key);

        String randomString = RandomUtils.randomString(16);
        //在接收消息的源头，加上分布式锁，避免集群部署，接收的消息重复处理
        if (null != redisDistributedLock) {
            boolean distLock = redisDistributedLock.lock(key, randomString, 20);
            if (!distLock) {
                log.info(" 获取分布式锁失败, 放弃处理.");
                return false;
            }
        }
        return true;
    }
}
